package com.sl.cdc.modular.debezium;

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.util.Map;

public class CdcFlinkSink extends RichSinkFunction<String> {
    private ObjectMapper mapper = new ObjectMapper();


    @Override
    public void invoke(String value, Context context) throws Exception {
        System.out.printf("数据发生变化: %s%n", value);
        TypeReference<Map<String, Object>> valueType = new TypeReference<Map<String, Object>>() {
        };
        Map<String, Object> result = mapper.readValue(value, valueType);
        Map<String, Object> payload = (Map<String, Object>) result.get("payload");
        String op = (String) payload.get("op");
        // 不对读操作处理
        if (!"r".equals(op)) {
//            MonitorMySQLCDC.queue.put(result);
            doAction(result);
        }
    }

    @SuppressWarnings("unchecked")
    private void doAction(Map<String, Object> result) throws Exception {
        Map<String, Object> payload = (Map<String, Object>) result.get("payload");
        String op = (String) payload.get("op");
        switch (op) {
            // 更新和插入操作
            case "u", "c" -> {
                Map<String, Object> after = (Map<String, Object>) payload.get("after");
                String id = after.get("id").toString();
                System.out.printf("操作：%s, ID: %s%n", op, id);
//                stringRedisTemplate.opsForValue().set(PREFIX + id, new ObjectMapper().writeValueAsString(after)) ;
            }
            // 删除操作
            case "d" -> {
                Map<String, Object> after = (Map<String, Object>) payload.get("before");
                String id = after.get("id").toString();
//                stringRedisTemplate.delete(PREFIX + id) ;
            }
        }
    }
}
